package com.yanfan.iot.ruleEngine;

import cn.hutool.core.util.HexUtil;
import com.yanfan.common.utils.StringUtils;
import com.yanfan.iot.model.MqttClientConfig;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
public class MqttBridgeClientFactory {

    private static final Map<Long, MqttAsyncClient> CLIENT_CACHE = new ConcurrentHashMap<>();
    private static final Map<Long, MqttClientConfig> CLIENT_CONFIG_CACHE = new ConcurrentHashMap<>();

    public static MqttAsyncClient instance(MqttClientConfig config) throws MqttException {
        if (config.isTemp()) {
            // 临时用的,不需要存,调用之后需要及时的关闭掉.
            return instanceNew(config);
        }

        if (config.getKey() == null) {
            return null;
        }

        MqttAsyncClient client = CLIENT_CACHE.get(config.getKey());
        MqttClientConfig oldConfig = CLIENT_CONFIG_CACHE.get(config.getKey());
        if (client == null) {
            client = instanceNew(config);
            CLIENT_CACHE.put(config.getKey(), client);
            CLIENT_CONFIG_CACHE.put(config.getKey(), config);
        } else if (!Objects.equals(oldConfig, config)) {
            if (client.isConnected()) {
                client.disconnect();
                client.close();
            }
            client = instanceNew(config);
            CLIENT_CACHE.put(config.getKey(), client);
            CLIENT_CONFIG_CACHE.put(config.getKey(), config);
        } else if (!client.isConnected()) {
            client.reconnect();
        }
        return client;
    }

    public static MqttAsyncClient instanceNew(MqttClientConfig config) {
        MqttAsyncClient client = null;
        try {
            client = new MqttAsyncClient(config.getHostUrl(), config.getClientId(), new MemoryPersistence());
            MqttConnectOptions options = new MqttConnectOptions();
            String username = config.getUsername();
            String password = config.getPassword();
            if (StringUtils.isNotEmpty(username)) {
                options.setUserName(username);
            }
            if (StringUtils.isNotEmpty(password)) {
                options.setPassword(password.toCharArray());
            }
            options.setCleanSession(true);
            options.setAutomaticReconnect(true);
            options.setMaxInflight(1000);
            // key 不为空, 如果为空的话表示是临时链接
            if (!config.isTemp()) {
                client.setCallback(new MqttCallback(client, options));
            }
            IMqttToken token = client.connect(options);
            token.waitForCompletion();
        } catch (MqttException e) {
            log.error("=>mqtt客户端创建错误，原因：" + e.getMessage());
        }
        return client;
    }

    public static void addSubscribe(Long key, String topic) {
        try {
            MqttAsyncClient client = CLIENT_CACHE.get(key);
            if (client != null) {
                log.warn("MqttAsyncClient：{}，topic：{}", key, topic);
                client.subscribe(topic, 0, new MessageArrivedHandler(key, topic));
            }
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    public static void addSubscribe(MqttAsyncClient client, Long key, String topic) {
        try {
            log.info("【MQTT数据桥接】 订阅 topic: {}", topic);
            client.subscribe(topic, 0, new MessageArrivedHandler(key, topic));
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    public static void delSubscribe(Long key, String topic) {
        try {
            MqttAsyncClient client = CLIENT_CACHE.get(key);
            if (client != null) {
                client.unsubscribe(topic);
            }
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    public static void remove(Long key) {
        MqttAsyncClient client = CLIENT_CACHE.get(key);
        if (client != null) {
            try {
                if (client.isConnected()) {
                    client.disconnect();
                    client.close();
                }
                CLIENT_CACHE.remove(key);
                CLIENT_CONFIG_CACHE.remove(key);
            } catch (MqttException e) {
                throw new RuntimeException(e);
            }
        }
    }

    public static class MqttCallback implements MqttCallbackExtended {

        /**
         * mqtt客户端
         */
        private MqttAsyncClient client;
        /**
         * 创建客户端参数
         */
        private MqttConnectOptions options;

        public MqttCallback(MqttAsyncClient client, MqttConnectOptions options) {
            this.client = client;
            this.options = options;
        }


        @Override
        public void connectComplete(boolean reconnect, String serverURI) {
            log.info("【MQTT数据桥接】 MQTT已经连接! serverURI: {}", serverURI);
        }

        @Override
        public void connectionLost(Throwable cause) {
            // 连接丢失后，一般在这里面进行重连
            log.debug("【MQTT数据桥接】 mqtt 连接丢失", cause);
            int count = 1;
            // int sleepTime = 0;
            boolean willConnect = true;
            while (willConnect) {
                try {
                    Thread.sleep(1000);
                    log.debug("【MQTT数据桥接】 连接[{}]断开，尝试重连第{}次", this.client.getServerURI(), count++);
                    if (this.client.isConnected()) {
                        this.client.reconnect();
                    } else {
                        this.client.connect(this.options);
                    }
                    log.debug("【MQTT数据桥接】 重连成功");
                    willConnect = false;
                } catch (Exception e) {
                    log.error("【MQTT数据桥接】 重连异常", e);
                }
            }
        }

        @Override
        public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
            // subscribe后得到的消息会执行到这里面
            log.info("【MQTT数据桥接】 订阅消息 topic: {}, msg: {}", topic, HexUtil.encodeHexStr(mqttMessage.getPayload()));
        }

        @Override
        public void deliveryComplete(IMqttDeliveryToken token) {
        }
    }
}
